package com.heima.kafka.stream;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.*;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author brianxia
 * @version 1.0
 * @date 2022-11-23 10:42
 */
public class KafkaStreamDemo {

    public static void main(String[] args) {
        //1.配置
        Properties properties = new Properties();
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"test109");

        //2.拓扑图 三要素 数据来源kafka topic   数据处理    数据输出 kafka topic
        StreamsBuilder streamsBuilder = new StreamsBuilder();
        KStream<String, String> source = streamsBuilder.stream("stream109_source");
        source
                //数据处理
                //hello kafka  将hello kafka用空格分割成多个数据 [hello ,kafka]
                .flatMapValues((ValueMapper<String, Iterable<String>>) value -> {
                    return Arrays.asList(value.split(" "));
                })
                //分组 hello * 2  kafka * 1 itcast * 1
                .groupBy((key, value) -> value)
                //设置时间窗口，每隔5秒钟处理一次
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                //统计个数
                .count()
                //将KTable转换成KStream
                .toStream()
                //将key value转换成字符串
                .map((KeyValueMapper<Windowed<String>, Long, KeyValue<String, String>>)
                        (objectWindowed, aLong) ->
                                new KeyValue<>(objectWindowed.key(),String.valueOf(aLong)))
                //输入到topic中
                .to("stream109_dest");


        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(),properties);

        kafkaStreams.start();
    }
}
